package com.young.sync.listener;

import com.young.canal.serializer.EventRecord;
import com.young.springboot.kafka.KafkaClientProperties;
import com.young.sync.config.TopicConf;
import com.young.sync.domain.ConsumeClientProperties;
import com.young.sync.factory.ConsumeListenerFactory;
import com.young.sync.factory.MessageListenerFactory;
import com.young.sync.service.entity.TopicEntity;
import com.young.sync.util.KafkaUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.List;

/**
 * @description：
 * @author: yangyahui01
 * @date: 12/5/24 3:23 PM
 */
@Component
@Slf4j
public class DataBusListener implements CommandLineRunner {


    @Autowired
    private TopicConf topicConf;

    @Autowired
    private KafkaUtils kafkaUtils;


    @Override
    public void run(String... args) throws Exception {

        initConsume();

        initTopic();

    }


    private void initConsume() {
        try {
            List<ConsumeClientProperties> consumes = topicConf.getConsumes();

            if (consumes == null || consumes.isEmpty()) {
                log.info("consume消费配置不存在或为空");
                return;
            }

            for (ConsumeClientProperties consumeClientProperties : consumes) {

                log.info("consumeClientProperties={}", consumeClientProperties);

                if (!ConsumeListenerFactory.isValidContainerListener(consumeClientProperties.getGroupId())) {
                    KafkaClientProperties kafkaClientProperties = buildKafkaClientProperties(consumeClientProperties);

                    ConcurrentKafkaListenerContainerFactory<String, EventRecord> containerFactory
                            = kafkaUtils.registerConsumeFactory(kafkaClientProperties);

                    ConsumeListenerFactory.register(consumeClientProperties.getGroupId(), containerFactory);
                }
            }
        } catch (Exception e) {
            log.error("consume消费添加失败:", e);
        }
    }

    /**
     * kafka消费者配置
     *
     * @param clientProperties
     * @return
     */
    private KafkaClientProperties buildKafkaClientProperties(ConsumeClientProperties clientProperties) {
        KafkaClientProperties kafkaClientProperties = new KafkaClientProperties();
        kafkaClientProperties.setBootstrapServers(clientProperties.getBootstrapServers());
        kafkaClientProperties.setEnableTransaction(clientProperties.isEnableTransaction());
        kafkaClientProperties.getConsumer().setGroupId(clientProperties.getGroupId());
        kafkaClientProperties.getConsumer().setAutoOffsetReset(clientProperties.getAutoOffsetReset());
        kafkaClientProperties.getListener().setAckMode(ContainerProperties.AckMode.valueOf(clientProperties.getAckMode()));
        kafkaClientProperties.getListener().setBatchListener(clientProperties.isBatchListener());
        kafkaClientProperties.getConsumer().setSessionTimeout(Duration.ofMillis(clientProperties.getSessionTimeout()));
        kafkaClientProperties.getProducer().setTransactionIdPrefix("tran-");
        kafkaClientProperties.buildAdminProperties();
        kafkaClientProperties.buildConsumerProperties();
        kafkaClientProperties.buildProducerProperties();
        kafkaClientProperties.getProperties();
        kafkaClientProperties.getConsumer()
                .setMaxPollRecords(clientProperties.getMaxPoll() != 0 ? clientProperties.getMaxPoll() : 1000);
        return kafkaClientProperties;
    }

    private void initTopic() {
        try {
            List<TopicEntity> topicList = topicConf.getTopicList();

            if (topicList == null || topicList.isEmpty()) {
                log.info("topic配置不存在或为空");
                return;
            }

            for (TopicEntity topic : topicList) {

                if (!Boolean.valueOf(topic.getIsConsume())) {
                    log.error("topic={}不消费", topic.getTopic());
                    continue;
                }

                try {
                    if (!MessageListenerFactory.isValidMessageListener(topic.getBizName())) {
                        kafkaUtils.registerMessageListener(topic);
                    }
                    MessageListenerFactory.register(topic.getBizName(), topic);
                } catch (Exception e) {
                    log.error("topic={}注册失败", topic.getTopic(), e);
                }
            }
        } catch (Exception e) {
            log.error("topic添加失败:", e);
        }
    }
}
